-
Notifications
You must be signed in to change notification settings - Fork 31
/
BlockingQueueAgent.fs
99 lines (79 loc) · 3.67 KB
/
BlockingQueueAgent.fs
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
// ----------------------------------------------------------------------------
// F# async extensions (BlockingQueueAgent.fs)
// (c) Tomas Petricek, 2011, Available under Apache 2.0 license.
// ----------------------------------------------------------------------------
namespace FSharpx.Control
open System
open System.Collections.Generic
// ----------------------------------------------------------------------------
type internal BlockingAgentMessage<'T> =
| AsyncAdd of 'T * AsyncReplyChannel<unit>
| Add of 'T
| AsyncGet of AsyncReplyChannel<'T>
/// Agent that implements an asynchronous queue with blocking put
/// and blocking get operation (this implements the producer-consumer
/// concurrent programming pattern). The constructor takes the maximal
/// size of the buffer.
type BlockingQueueAgent<'T>(maxLength) =
[<VolatileField>]
let mutable count = 0
let agent = Agent.Start(fun agent ->
let queue = new Queue<'T>()
let rec emptyQueue() =
agent.Scan(fun msg ->
match msg with
| AsyncAdd(value, reply) -> Some(enqueueAndContinueWithReply(value, reply))
| Add(value) -> Some(enqueueAndContinue(value))
| _ -> None )
and fullQueue() =
agent.Scan(fun msg ->
match msg with
| AsyncGet(reply) -> Some(dequeueAndContinue(reply))
| _ -> None )
and runningQueue() = async {
let! msg = agent.Receive()
match msg with
| AsyncAdd(value, reply) -> return! enqueueAndContinueWithReply(value, reply)
| Add(value) -> return! enqueueAndContinue(value)
| AsyncGet(reply) -> return! dequeueAndContinue(reply) }
and enqueueAndContinueWithReply (value, reply) = async {
reply.Reply()
queue.Enqueue(value)
count <- queue.Count
return! chooseState() }
and enqueueAndContinue (value) = async {
queue.Enqueue(value)
count <- queue.Count
return! chooseState() }
and dequeueAndContinue (reply) = async {
reply.Reply(queue.Dequeue())
count <- queue.Count
return! chooseState() }
and chooseState() =
if queue.Count = 0 then emptyQueue()
elif queue.Count < maxLength then runningQueue()
else fullQueue()
// Start with an empty queue
emptyQueue() )
/// Asynchronously adds item to the queue. The operation ends when
/// there is a place for the item. If the queue is full, the operation
/// will block until some items are removed.
member x.AsyncAdd(v:'T, ?timeout) =
agent.PostAndAsyncReply((fun ch -> AsyncAdd(v, ch)), ?timeout=timeout)
/// Asynchronously adds item to the queue. The operation ends when
/// there is a place for the item. If the queue is full, the operation
/// will block until some items are removed. This overload does not
/// signal the caller that the item has been added.
member x.Add(v:'T) =
agent.Post(Add v)
/// Asynchronously gets item from the queue. If there are no items
/// in the queue, the operation will block until items are added.
member x.AsyncGet(?timeout) =
agent.PostAndAsyncReply(AsyncGet, ?timeout=timeout)
/// Synchronously gets item from the queue. If there are no items
/// in the queue, the operation will block until items are added.
/// This method blocks until value is available!
member x.Get(?timeout) =
agent.PostAndReply(AsyncGet, ?timeout=timeout)
/// Gets the number of elements currently waiting in the queue.
member x.Count = count